package worker

import (
	"bufio"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"io/ioutil"
	"math/rand"
	"os"
	"os/exec"
	"path/filepath"
	"recorder/control"
	"recorder/internal"
	"recorder/log"
	"strings"
	"syscall"
	"time"
)

const (
	reconnectTimeout = 1 * time.Second
)

type Worker struct {
	cfg    *internal.Config
	done   chan struct{}
	client *control.Client
}

func New(config *internal.Config) *Worker {
	return &Worker{
		cfg:    config,
		done:   make(chan struct{}),
		client: control.NewClient(config),
	}
}

func (w *Worker) Close() {
	log.Info("worker exiting")
	close(w.done)
}

func (w *Worker) handleConn(c *control.Conn) {
	defer c.Close()
	for {
		log.Info("worker %s wait for job xxxxxx", w.cfg.Id)
		var j *internal.Job = nil
		select {
		case <-w.done:
			log.Info("worker %s closed while waiting for job", w.cfg.Id)
			return
		case <-c.Dis:
			log.Info("worker %s disconnected while waiting for job", w.cfg.Id)
			return
		case m := <-c.In:
			log.Info("worker %s received %v", w.cfg.Id, m)
			if m == nil || m.Action != control.ActionRun {
				return
			}
			j = m.Job
		}

		select {
		case <-w.done:
			log.Warning("worker %s closed while sending ack", w.cfg.Id)
			return
		case <-c.Dis:
			log.Warning("worker %s disconnected while sending ack", w.cfg.Id)
			return
		case c.Out <- &control.Message{Action: control.ActionAck}:
			log.Info("worker %s send ack while sending ack", w.cfg.Id)
		}

		jobOutFile, jobErr := w.executeJob_2(c.Dis, j)
		resultMsg := &control.Message{
			Action: control.ActionDone,
			Result: &internal.Result{
				Job:   j.Scene,
				Error: "",
			},
		}
		if jobErr != nil {
			resultMsg.Result.Error = jobErr.Error()
		}
		select {
		case <-w.done:
			log.Error("worker %s closed while sending done", w.cfg.Id)
			if len(jobOutFile) != 0 {
				os.Remove(jobOutFile)
			}
			return
		case <-c.Dis:
			log.Error("worker %s disconnected while sending done", w.cfg.Id)
			if len(jobOutFile) != 0 {
				os.Remove(jobOutFile)
			}
			return
		case c.Out <- resultMsg:
		}

		select {
		case <-w.done:
			log.Error("worker %s closed while waiting for ack", w.cfg.Id)
			if len(jobOutFile) != 0 {
				os.Remove(jobOutFile)
			}
			return
		case <-c.Dis:
			log.Error("worker %s disconnected while waiting for ack", w.cfg.Id)
			if len(jobOutFile) != 0 {
				os.Remove(jobOutFile)
			}
			return
		case m := <-c.In:
			if m == nil || m.Action != control.ActionAck {
				log.Info("worker %s waiting for ack, received %+v", w.cfg.Id, m)
				if len(jobOutFile) != 0 {
					os.Remove(jobOutFile)
				}
				return
			}
			w.outputFiles(j, jobOutFile, jobErr)
		}
	}
}

func (w *Worker) Run() {
	log.Info("worker started")
	go w.client.Connect()
	go w.cleanup()
	for {
		// read a job
		select {
		case <-w.done:
			log.Info("worker closed")
			return
		case c := <-w.client.C:
			w.handleConn(c)
			select {
			case <-w.done:
				return
			default:
			}
			go w.client.Connect()
		}
	}
}

// todo: harvest this goroutine when exiting
func (w *Worker) cleanup() {
	timer := time.NewTimer((30 + time.Duration(rand.Int()%31)) * time.Minute)
	defer timer.Stop()
	for {
		select {
		case now := <-timer.C:
			before := now.Add(time.Duration(w.cfg.KeepDays*-24) * time.Hour)
			w.cleanupDir(w.cfg.BakPath, before)
			w.cleanupDir(w.cfg.OutPath, before)
			w.cleanupDir(w.cfg.LogPath, before)
			w.cleanupDir(w.cfg.TmpPath, before)
			timer.Reset((30 + time.Duration(rand.Int()%31)) * time.Minute)
		case <-w.done:
			return
		}
	}
}

func (w *Worker) executeJob(done chan struct{}, j *internal.Job_bak) (string, error) {
	type MergeInput struct {
		file   string
		source string
	}
	log.Info("%s: begin step 1", j.Scene)

	logWriter := NewLogWriter(j.Scene, filepath.Join(w.cfg.LogPath, "detail"))

	merge_in_files := make([]*MergeInput, 0)
	for src_idx, src := range j.Sources {
		rtcList := make([]string, 0)
		segmentsByRtc := make(map[string][]*internal.Segment)

		j.Sources[src_idx].Scene = j.Scene

		for seg_idx, seg := range src.Segments {
			rtc := seg.Rtc
			rtcExist := false
			for _, r := range rtcList {
				if r == rtc {
					rtcExist = true
					break
				}
			}
			if !rtcExist {
				rtcList = append(rtcList, rtc)
			}
			j.Sources[src_idx].Segments[seg_idx].Scene = j.Scene
			j.Sources[src_idx].Segments[seg_idx].Source = j.Sources[src_idx].Name

			_, ok := segmentsByRtc[rtc]
			if !ok {
				segmentsByRtc[rtc] = make([]*internal.Segment, 0)
			}
			segmentsByRtc[rtc] = append(segmentsByRtc[rtc], j.Sources[src_idx].Segments[seg_idx])
		}

		for _, rtc := range rtcList {
			segments := segmentsByRtc[rtc]
			transfmt_output, err := w.doTransfmt(done, j.Scene, j.Date, j.Time, rtc, segments, logWriter)
			select {
			case <-done:
				log.Warning("job %v aborted", j.Scene)
				return "", errors.New("cancelled")
			default:
			}
			if err != nil {
				log.Warning("%v: %v transfmt err=[%v]", j.Scene, rtc, err)
				continue
			}

			merge_in_files = append(merge_in_files, &MergeInput{
				file:   transfmt_output,
				source: src.Name,
			})
		}
	}

	extension_name := "mp4"
	if j.Type == "picture" {
		extension_name = "jpg"
	}
	switch {
	case len(merge_in_files) == 0:
		log.Error("%v: error: 0 files to be merged", j.Scene)
		return "", errors.New("noinput")
	case len(merge_in_files) == 1 && w.cfg.NotMergeWhenOneFile:
		if j.Type == "movie" {
			f := merge_in_files[0].file
			defer os.Remove(fmt.Sprintf("%v.ts", f[0:len(f)-len(filepath.Ext(f))]))
			return f, nil
		} else {
			in := merge_in_files[0].file
			in_ts := fmt.Sprintf("%v.ts", in[0:len(in)-len(filepath.Ext(in))])
			out := fmt.Sprintf("%v.jpg", in[0:len(in)-len(filepath.Ext(in))])
			defer os.Remove(in)
			defer os.Remove(in_ts)
			args := make([]string, 0)
			args = append(args, "-max_error_rate", "1.0", "-y", "-i", in, "-an", "-vf", "select='eq(pict_type\\,I)'", "-vframes", "1", "-c:v", "mjpeg", "-f", "mjpeg", "-q:v", "1", out)
			log.Info("%v: picture: run [ffmpeg %v]", j.Scene, strings.Trim(fmt.Sprint(args), "[]"))
			cmd := exec.Command("ffmpeg", args...)
			cmd.SysProcAttr = &syscall.SysProcAttr{
				Setpgid: true,
			}
			cmd.Stdout = logWriter
			cmd.Stderr = logWriter
			err := cmd.Start()
			if err != nil {
				log.Error("%v: picture: run ffmpeg err=[%v]", j.Scene, err)
				os.Remove(out)
				return "", err
			}
			pictureDone := make(chan error)
			go func() {
				pictureDone <- cmd.Wait()
			}()

			select {
			case <-done:
				log.Warning("job %v: aborting", j.Scene)
				cmd.Process.Kill()
				<-pictureDone
				log.Warning("job %v: aborted", j.Scene)
				return "", errors.New("cancelled")
			case err = <-pictureDone:
			}

			if err != nil {
				log.Error("job %v: picture: wait err=[%v]", j.Scene, err)
				os.Remove(out)
				return "", err
			}

			log.Info("job %v: picture ok", j.Scene)
			return out, nil
		}
	default:
		log.Info("%v: begin merge", j.Scene)
		merge_out := filepath.Join(w.cfg.TmpPath, j.Date, j.Time, j.Scene, fmt.Sprintf("%v.%v", j.Scene, extension_name))
		os.MkdirAll(filepath.Dir(merge_out), os.ModePerm)
		args := make([]string, 0)
		last_source := ""
		for _, mi := range merge_in_files {
			if mi.source == last_source {
				args = append(args, "-con")
			}
			last_source = mi.source
			args = append(args, "-i", mi.file)
			//添加水印，事例
			//args = append(args, "--water-mark", "movie='/Users/yangss/Downloads/xswitch_samples/logo.png':stream_index=0;overlay=main_w-overlay_w-5:5")
			defer os.Remove(mi.file)
			defer os.Remove(fmt.Sprintf("%v.ts", mi.file[0:len(mi.file)-len(filepath.Ext(mi.file))]))
		}
		if j.Type == "picture" {
			args = append(args, "-max_error_rate", "1.0")
			args = append(args, "-an", "-ss", "00:00:01", "-vframes", "1", "-c:v", "mjpeg", "-f", "mjpeg", "-q:v", "1", merge_out)
		} else {
			args = append(args, "-c:v", "libx264", "-preset", "superfast", "-crf", "28", "-r", "15", "-movflags", "+faststart", "-c:a", "aac", "-ar", "48000", "-ac", "2", merge_out)
			// args = append(args, "-c:v", "libx264", "-preset", "superfast", "-r", "15", "-movflags", "+faststart", "-c:a", "aac", "-ar", "48000", "-ac", "2", merge_out)
		}

		log.Info("%v: exec=[merge %v]", j.Scene, strings.Trim(fmt.Sprint(args), "[]"))
		cmd := exec.Command("merge", args...)
		cmd.SysProcAttr = &syscall.SysProcAttr{
			Setpgid: true,
		}
		cmd.Stdout = logWriter
		cmd.Stderr = logWriter
		err := cmd.Start()
		if err != nil {
			log.Error("%v: exec err=[%v]", j.Scene, err)
			os.Remove(merge_out)
			return "", err
		}

		mergeDone := make(chan error)
		go func() {
			mergeDone <- cmd.Wait()
		}()

		select {
		case <-done:
			log.Warning("job %v aborting", j.Scene)
			cmd.Process.Kill()
			<-mergeDone
			os.Remove(merge_out)
			log.Warning("job %v aborted", j.Scene)
			return "", errors.New("cancelled")
		case err = <-mergeDone:
		}

		if err != nil {
			log.Error("job %v: merge child err=[%v]", j.Scene, err)
			os.Remove(merge_out)
			return "", err
		}

		log.Info("%v: merge ok", j.Scene)
		return merge_out, nil
	}
}

func (w *Worker) executeJob_2(done chan struct{}, j *internal.Job) (string, error) {

	type MergeInput struct {
		file   string
		source string
	}
	log.Info("%s: begin step 1", j.Scene)

	logWriter := NewLogWriter(j.Scene, filepath.Join(w.cfg.LogPath, "detail"))
	i := 0
	sourceVideos := make([]string, 0)
	extension_name := "mp4"

	for _, sourceMgr := range j.SourceList {
		i++
		merge_in_files := make([]*MergeInput, 0)
		segmentsByRtc := make(map[string]*internal.Segment)

		for _, source := range sourceMgr.Sources {
			source.Scene = j.Scene
			source.Segment.Scene = j.Scene
			source.Segment.Source = source.Name
			segmentsByRtc[source.Segment.Rtc] = &source.Segment
		}

		segMap, _ := json.Marshal(segmentsByRtc)
		log.Info("executeJob_2.segmentsByRtc===%s", string(segMap))

		for rtc, segment := range segmentsByRtc {
			transfmt_output, err := w.doTransfmt_2(done, j.Scene, j.Date, j.Time, rtc, segment, logWriter)
			select {
			case <-done:
				log.Warning("job %v aborted", j.Scene)
				return "", errors.New("cancelled")
			default:
			}
			if err != nil {
				log.Warning("%v: %v transfmt err=[%v]", j.Scene, rtc, err)
				continue
			}

			merge_in_files = append(merge_in_files, &MergeInput{
				file:   transfmt_output,
				source: segment.Source,
			})
			segmentsByRtc = make(map[string]*internal.Segment)
		}

		if j.Type == "picture" {
			extension_name = "jpg"
		}
		switch {
		case len(merge_in_files) == 0:
			log.Error("%v: error: 0 files to be merged", j.Scene)
			return "", errors.New("noinput")
		case len(merge_in_files) == 1 && w.cfg.NotMergeWhenOneFile:
			if j.Type == "movie" {
				f := merge_in_files[0].file
				defer os.Remove(fmt.Sprintf("%v.ts", f[0:len(f)-len(filepath.Ext(f))]))
				return f, nil
			} else {
				in := merge_in_files[0].file
				in_ts := fmt.Sprintf("%v.ts", in[0:len(in)-len(filepath.Ext(in))])
				out := fmt.Sprintf("%v.jpg", in[0:len(in)-len(filepath.Ext(in))])
				defer os.Remove(in)
				defer os.Remove(in_ts)
				args := make([]string, 0)
				args = append(args, "-max_error_rate", "1.0", "-y", "-i", in, "-an", "-vf", "select='eq(pict_type\\,I)'", "-vframes", "1", "-c:v", "mjpeg", "-f", "mjpeg", "-q:v", "1", out)
				log.Info("%v: picture: run [ffmpeg %v]", j.Scene, strings.Trim(fmt.Sprint(args), "[]"))
				cmd := exec.Command("ffmpeg", args...)
				cmd.SysProcAttr = &syscall.SysProcAttr{
					Setpgid: true,
				}
				cmd.Stdout = logWriter
				cmd.Stderr = logWriter
				err := cmd.Start()
				if err != nil {
					log.Error("%v: picture: run ffmpeg err=[%v]", j.Scene, err)
					os.Remove(out)
					return "", err
				}
				pictureDone := make(chan error)
				go func() {
					pictureDone <- cmd.Wait()
				}()

				select {
				case <-done:
					log.Warning("job %v: aborting", j.Scene)
					cmd.Process.Kill()
					<-pictureDone
					log.Warning("job %v: aborted", j.Scene)
					return "", errors.New("cancelled")
				case err = <-pictureDone:
				}

				if err != nil {
					log.Error("job %v: picture: wait err=[%v]", j.Scene, err)
					os.Remove(out)
					return "", err
				}

				log.Info("job %v: picture ok", j.Scene)
				return out, nil
			}
		default:
			log.Info("%v: begin merge", j.Scene)
			merge_out := filepath.Join(w.cfg.TmpPath, j.Date, j.Time, j.Scene, fmt.Sprintf("%v-%v.%v", j.Scene, i, extension_name))
			os.MkdirAll(filepath.Dir(merge_out), os.ModePerm)
			sourceVideos = append(sourceVideos, fmt.Sprintf("%v-%v.%v", j.Scene, i, extension_name))
			args := make([]string, 0)
			last_source := ""
			for _, mi := range merge_in_files {
				if mi.source == last_source {
					args = append(args, "-con")
				}
				last_source = mi.source
				args = append(args, "-i", mi.file)
				//添加水印示例
				//args = append(args, "--water-mark", "movie='/Users/yangss/Downloads/xswitch_samples/logo.png':stream_index=0;overlay=main_w-overlay_w-5:5")
				defer os.Remove(mi.file)
				defer os.Remove(fmt.Sprintf("%v.ts", mi.file[0:len(mi.file)-len(filepath.Ext(mi.file))]))
			}
			if j.Type == "picture" {
				args = append(args, "-max_error_rate", "1.0")
				args = append(args, "-an", "-ss", "00:00:01", "-vframes", "1", "-c:v", "mjpeg", "-f", "mjpeg", "-q:v", "1", merge_out)
			} else {
				args = append(args, "-c:v", "libx264", "-preset", "superfast", "-crf", "28", "-r", "15", "-movflags", "+faststart", "-c:a", "aac", "-ar", "48000", "-ac", "2", merge_out)
				// args = append(args, "-c:v", "libx264", "-preset", "superfast", "-r", "15", "-movflags", "+faststart", "-c:a", "aac", "-ar", "48000", "-ac", "2", merge_out)
			}

			log.Info("%v: exec=[merge %v]", j.Scene, strings.Trim(fmt.Sprint(args), "[]"))
			cmd := exec.Command("merge", args...)
			cmd.SysProcAttr = &syscall.SysProcAttr{
				Setpgid: true,
			}
			cmd.Stdout = logWriter
			cmd.Stderr = logWriter
			err := cmd.Start()
			if err != nil {
				log.Error("%v: exec err=[%v]", j.Scene, err)
				os.Remove(merge_out)
				return "", err
			}

			mergeDone := make(chan error)
			go func() {
				mergeDone <- cmd.Wait()
			}()

			select {
			case <-done:
				log.Warning("job %v aborting", j.Scene)
				cmd.Process.Kill()
				<-mergeDone
				os.Remove(merge_out)
				log.Warning("job %v aborted", j.Scene)
				return "", errors.New("cancelled")
			case err = <-mergeDone:
			}

			if err != nil {
				log.Error("job %v: merge child err=[%v]", j.Scene, err)
				os.Remove(merge_out)
				return "", err
			}
			log.Info("%v: merge ok", j.Scene)
		}
	}
	// 写文件
	senceTxt := filepath.Join(w.cfg.TmpPath, j.Date, j.Time, j.Scene, fmt.Sprintf("%v.%v", j.Scene, "txt"))
	err := writeTxt(senceTxt, sourceVideos)
	if err != nil {
		log.Warning("writeTxt open file error", err)
		return "", err
	}
	log.Info("%v: write file success", j.Scene)
	// 拼接 ffmpeg -f concat -i lzbffmpeg\videos.txt joinoutput.mp4
	return w.SplicingVideo(done, j, extension_name, logWriter)
}

func writeTxt(filepath string, sourceVideos []string) error {
	f, err := os.OpenFile(filepath, os.O_CREATE|os.O_WRONLY, 0)
	if err != nil {
		log.Warning("writeTxt open file error", err)
		return err
	}
	defer f.Close()
	writer := bufio.NewWriter(f)
	for _, v := range sourceVideos {
		text := fmt.Sprintf("%v '%v'", "file", v)
		_, err := writer.WriteString(text + "\r\n")
		if err != nil {
			log.Warning("writeTxt write txt error", err)
			return err
		}
	}
	writer.Flush()
	return nil
}

func (w Worker) SplicingVideo(done chan struct{}, j *internal.Job, extension_name string, logWriter *LogWriter) (string, error) {
	videoPath := filepath.Join(w.cfg.TmpPath, j.Date, j.Time, j.Scene, fmt.Sprintf("%v.%v", j.Scene, extension_name))
	os.MkdirAll(filepath.Dir(videoPath), os.ModePerm)
	command := fmt.Sprintf("-f concat -i %s %s", fmt.Sprintf("%v.%v", j.Scene, "txt"), fmt.Sprintf("%v.%v", j.Scene, extension_name))
	log.Info("SplicingVideo the cmd is:%s", command)

	cmd := exec.Command("ffmpeg", command)
	cmd.SysProcAttr = &syscall.SysProcAttr{
		Setpgid: true,
	}
	cmd.Stdout = logWriter
	cmd.Stderr = logWriter
	err := cmd.Start()
	if err != nil {
		log.Error("%v: splicing video exec err=[%v]", j.Scene, err)
		//os.Remove(videoPath)
		return "", err
	}
	mergeDone := make(chan error)
	go func() {
		mergeDone <- cmd.Wait()
	}()

	select {
	case <-done:
		log.Warning("splicing video job %v aborting", j.Scene)
		cmd.Process.Kill()
		<-mergeDone
		//os.Remove(videoPath)
		log.Warning("splicing video job %v aborted", j.Scene)
		return "", errors.New("cancelled")
	case err = <-mergeDone:
	}

	if err != nil {
		log.Error("splicing video job %v: merge child err=[%v]", j.Scene, err)
		// os.Remove(videoPath)
		return "", err
	}
	log.Info("%v: vertical merge success", j.Scene)
	return videoPath, nil
}

func (w *Worker) doTransfmt_2(done chan struct{}, sceneName, sceneDate, sceneTime, rtcName string, segment *internal.Segment, logWriter *LogWriter) (string, error) {
	tlv := filepath.Join(w.cfg.TlvPath, rtcName)
	log.Info("%v: begin transfmt tlv=[%v]", sceneName, rtcName)
	defer log.Info("%v: end transfmt tlv=[%v]", sceneName, rtcName)
	if segment == nil {
		log.Warning("%v: no segment", sceneName)
		return "", errors.New("no segment")
	}
	out_pathname := filepath.Join(w.cfg.TmpPath, sceneDate, sceneTime, sceneName, filepath.Base(rtcName))
	out_pathname = out_pathname[0 : len(out_pathname)-len(filepath.Ext(rtcName))]
	err := os.MkdirAll(filepath.Dir(out_pathname), os.ModePerm)
	if err != nil {
		log.Error("create out_pathname fail err = %v", err)
	}
	args := []string{"-i", tlv, "-o", out_pathname}
	if !segment.Audio {
		args = append(args, "-A")
	}
	if !segment.Video {
		args = append(args, "-V")
	}
	log.Info("%v: add segment=[%v]", sceneName, segment)
	args = append(args, "-f", fmt.Sprint(segment.From), "-t", fmt.Sprint(segment.To))
	log.Info("%v: %v: exec=[transfmt %v]", sceneName, rtcName, strings.Trim(fmt.Sprint(args), "[]"))

	cmd := exec.Command("transfmt", args...)
	cmd.SysProcAttr = &syscall.SysProcAttr{
		Setpgid: true,
	}
	cmd.Stdout = logWriter
	stderr, err := cmd.StderrPipe()
	if err != nil {
		log.Error("%v: %v: open transfmt stderr err=[%v]", sceneName, rtcName, err)
		return "", err
	}
	err = cmd.Start()
	if err != nil {
		log.Error("job %v: tlv %v: exec transfmt err=[%v]", sceneName, rtcName, err)
		return "", err
	}
	type transfmtResult struct {
		File string
		Err  error
	}
	var result *transfmtResult
	transfmtDone := make(chan *transfmtResult)
	go func() {
		var output_file string
		stderr_out, err := ioutil.ReadAll(stderr)
		if err != nil {
			log.Error("job %v: tlv %v: read transfmt stderr err=[%v]", sceneName, rtcName, err)
		} else {
			logWriter.Write(stderr_out)
			out_str := string(stderr_out)
			lines := strings.Split(out_str, "\n")
			for _, line := range lines {
				if strings.HasPrefix(line, "output filename = ") {
					parts := strings.Split(line, "output filename = ")
					if len(parts) == 2 {
						output_file = parts[1]
					}
				}
			}
		}
		err = cmd.Wait()
		if err != nil {
			log.Error("job %v: tlv %v: transfmt child error %v", sceneName, rtcName, err)
		}
		transfmtDone <- &transfmtResult{
			File: output_file,
			Err:  err,
		}
	}()

	select {
	case <-done:
		log.Warning("job %v: tlv %v: aborting", sceneName, rtcName)
		cmd.Process.Kill()
		result = <-transfmtDone
		if len(result.File) != 0 {
			os.Remove(result.File)
		}
		log.Warning("job %v: tlv %v: aborted", sceneName, rtcName)
		return "", errors.New("cancelled")
	case result = <-transfmtDone:
	}

	if result.Err != nil {
		log.Error("job %v: tlv %v: transfmt child error: %v", sceneName, rtcName, result.Err)
		if len(result.File) != 0 {
			os.Remove(result.File)
		}
		return "", result.Err
	}

	if len(result.File) == 0 {
		log.Error("job %v: tlv %s: cannot get output filename from transfmt", sceneName, rtcName)
		return "", errors.New("transfmt no output")
	}
	log.Info("job %v: output filename: [%v]", sceneName, result.File)
	return result.File, nil
}

func (w *Worker) doTransfmt(done chan struct{}, sceneName, sceneDate, sceneTime, rtcName string, segments []*internal.Segment, logWriter *LogWriter) (string, error) {
	tlv := filepath.Join(w.cfg.TlvPath, rtcName)
	log.Info("%v: begin transfmt tlv=[%v]", sceneName, rtcName)
	defer log.Info("%v: end transfmt tlv=[%v]", sceneName, rtcName)
	if len(segments) == 0 {
		log.Warning("%v: no segment", sceneName)
		return "", errors.New("no segment")
	}
	out_pathname := filepath.Join(w.cfg.TmpPath, sceneDate, sceneTime, sceneName, filepath.Base(rtcName))
	out_pathname = out_pathname[0 : len(out_pathname)-len(filepath.Ext(rtcName))]
	os.MkdirAll(filepath.Dir(out_pathname), os.ModePerm)
	args := []string{
		"-i",
		tlv,
		"-o",
		out_pathname,
	}
	if !segments[0].Audio {
		args = append(args, "-A")
	}
	if !segments[0].Video {
		args = append(args, "-V")
	}

	for _, s := range segments {
		log.Info("%v: add segment=[%v]", sceneName, s)
		args = append(args, "-f", fmt.Sprint(s.From), "-t", fmt.Sprint(s.To))
	}
	log.Info("%v: %v: exec=[transfmt %v]", sceneName, rtcName, strings.Trim(fmt.Sprint(args), "[]"))
	cmd := exec.Command("transfmt", args...)
	cmd.SysProcAttr = &syscall.SysProcAttr{
		Setpgid: true,
	}
	cmd.Stdout = logWriter
	stderr, err := cmd.StderrPipe()
	if err != nil {
		log.Error("%v: %v: open transfmt stderr err=[%v]", sceneName, rtcName, err)
		return "", err
	}
	err = cmd.Start()
	if err != nil {
		log.Error("job %v: tlv %v: exec transfmt err=[%v]", sceneName, rtcName, err)
		return "", err
	}
	type transfmtResult struct {
		File string
		Err  error
	}
	var result *transfmtResult
	transfmtDone := make(chan *transfmtResult)
	go func() {
		var output_file string
		stderr_out, err := ioutil.ReadAll(stderr)
		if err != nil {
			log.Error("job %v: tlv %v: read transfmt stderr err=[%v]", sceneName, rtcName, err)
		} else {
			logWriter.Write(stderr_out)
			out_str := string(stderr_out)
			lines := strings.Split(out_str, "\n")
			for _, line := range lines {
				if strings.HasPrefix(line, "output filename = ") {
					parts := strings.Split(line, "output filename = ")
					if len(parts) == 2 {
						output_file = parts[1]
					}
				}
			}
		}
		err = cmd.Wait()
		if err != nil {
			log.Error("job %v: tlv %v: transfmt child error %v", sceneName, rtcName, err)
		}
		transfmtDone <- &transfmtResult{
			File: output_file,
			Err:  err,
		}
	}()

	select {
	case <-done:
		log.Warning("job %v: tlv %v: aborting", sceneName, rtcName)
		cmd.Process.Kill()
		result = <-transfmtDone
		if len(result.File) != 0 {
			os.Remove(result.File)
		}
		log.Warning("job %v: tlv %v: aborted", sceneName, rtcName)
		return "", errors.New("cancelled")
	case result = <-transfmtDone:
	}

	if result.Err != nil {
		log.Error("job %v: tlv %v: transfmt child error: %v", sceneName, rtcName, result.Err)
		if len(result.File) != 0 {
			os.Remove(result.File)
		}
		return "", result.Err
	}

	if len(result.File) == 0 {
		log.Error("job %v: tlv %s: cannot get output filename from transfmt", sceneName, rtcName)
		return "", errors.New("transfmt no output")
	}
	log.Info("job %v: output filename: [%v]", sceneName, result.File)
	return result.File, nil
}

func (w *Worker) outputFiles(j *internal.Job, tempMediaFile string, err error) {
	// var ext_name string
	ext_name := "mp3"

	if _, err := os.Stat(tempMediaFile); err != nil {
		tempMediaFile = fmt.Sprintf("%v/%v.%v", filepath.Dir(tempMediaFile), j.Scene, ext_name)

		if _, err = os.Stat(tempMediaFile); err != nil {
			log.Error("job %v find err=[%v]", j.Scene, err)
			return
		}

		log.Info("job %v file extension change to mp3, tmpmedia=[%v]", j.Scene, tempMediaFile)
	}

	log.Info("job %v done", j.Scene)
	if err == nil {
		lockOutMediaDir := filepath.Join(w.cfg.OutPath, fmt.Sprintf("%v.lock", j.SubDir))
		finalOutMediaDir := filepath.Join(w.cfg.OutPath, j.SubDir)

		count := uint16(0)
		for err = os.MkdirAll(lockOutMediaDir, os.ModePerm); err != nil && count < 100; { //创建失败，继续创建
			log.Error("job %v create lockdir=[%v] err=[%v] count=[%d]", j.Scene, lockOutMediaDir, err, count)

			count++
			err = os.MkdirAll(lockOutMediaDir, os.ModePerm)
		}
		if err != nil {
			log.Info("job %v create lock dir fail. err=[%v], [%v]", j.Scene, err, lockOutMediaDir)
			return
		}

		outMediaFile := filepath.Join(lockOutMediaDir, fmt.Sprintf("%v%v", j.Scene, filepath.Ext(tempMediaFile)))
		//os.MkdirAll(filepath.Dir(outMediaFile), os.ModePerm)
		if err := os.Rename(tempMediaFile, outMediaFile); err != nil {
			log.Info("job %v rename mediafile error, fallback to copy, err=[%v] : [%v] -> [%v]", j.Scene, err, tempMediaFile, outMediaFile)
			err = copyFileContents(tempMediaFile, outMediaFile)
			os.Remove(tempMediaFile)
		}
		if err != nil {
			log.Info("job %v copy mediafile fail. err=[%v] : [%v] -> [&v]", j.Scene, err, tempMediaFile, outMediaFile)
			return
		} else {
			log.Info("job %v copy mediafile success. [%v] -> [%v]", j.Scene, tempMediaFile, outMediaFile)
		}

		tempMediaJson := tempMediaFile[0:len(tempMediaFile)-len(filepath.Ext(tempMediaFile))] + ".ffprobe.json"
		outMediaJson := filepath.Join(lockOutMediaDir, fmt.Sprintf("%v.ffprobe.json", j.Scene))
		if err := os.Rename(tempMediaJson, outMediaJson); err != nil {
			log.Info("job %v rename jsonfile error, fallback to copy, err=[%v] : [%v] -> [%v]", j.Scene, err, tempMediaJson, outMediaJson)
			err = copyFileContents(tempMediaJson, outMediaJson)
			os.Remove(tempMediaJson)
		}
		if err != nil {
			log.Info("job %v copy jsonfile fail. err=[%v] : [%v] -> [&v]", j.Scene, err, tempMediaJson, outMediaJson)
			return
		} else {
			log.Info("job %v copy jsonfile success. [%v] -> [%v]", j.Scene, tempMediaJson, outMediaJson)
		}

		tempCtx := filepath.Join(lockOutMediaDir, fmt.Sprintf("%v.ctx.tmp", j.Scene))
		outCtx := filepath.Join(lockOutMediaDir, fmt.Sprintf("%v.ctx", j.Scene))

		if err == nil {
			if err := ioutil.WriteFile(tempCtx, []byte(j.Ctx), os.ModePerm); err != nil {
				log.Warning("job %v write tmp ctx err=[%v], try again", j.Scene, err)
				if err := ioutil.WriteFile(tempCtx, []byte(j.Ctx), os.ModePerm); err != nil {
					log.Warning("job %v write tmp ctx err=[%v], aborted", j.Scene, err)
				}
			}
		}
		if err == nil {
			if err = os.Rename(tempCtx, outCtx); err != nil {
				log.Warning("job %v rename tmp ctx err=[%v], try again", j.Scene, err)
				if err = os.Rename(tempCtx, outCtx); err != nil {
					log.Warning("job %v rename tmp ctx err=[%v], aborted", j.Scene, err)
				}
			}
		}
		if err == nil {
			if err = os.Rename(lockOutMediaDir, finalOutMediaDir); err != nil {
				log.Warning("job %v rename lockdir err=[%v], try copy", j.Scene, err)
				err = copyFileContents(outMediaFile, filepath.Join(finalOutMediaDir, fmt.Sprintf("%v%v", j.Scene, filepath.Ext(tempMediaFile))))
				if err == nil {
					err = copyFileContents(outCtx, filepath.Join(finalOutMediaDir, fmt.Sprintf("%v.ctx", j.Scene)))
					err = copyFileContents(outMediaJson, filepath.Join(finalOutMediaDir, fmt.Sprintf("%v.ffprobe.json", j.Scene)))
				}

				os.RemoveAll(lockOutMediaDir)
				log.Info("job %v remove all [%v]", j.Scene, lockOutMediaDir)
			}
		}

		if err != nil {
			log.Info("job %v file operate fail. err=[%v], ctx=[%v]", j.Scene, err, outCtx)
		} else {
			log.Info("job %v, success ctx in [%v]", j.Scene, finalOutMediaDir)
		}
	}
}

// copyFileContents copies the contents of the file named src to the file named
// by dst. The file will be created if it does not already exist. If the
// destination file exists, all it's contents will be replaced by the contents
// of the source file.
func copyFileContents(src, dst string) (err error) {
	in, err := os.Open(src)
	if err != nil {
		return
	}
	defer in.Close()
	out, err := os.Create(dst)
	if err != nil {
		return
	}
	defer func() {
		cerr := out.Close()
		if err == nil {
			err = cerr
		}
	}()
	if _, err = io.Copy(out, in); err != nil {
		return
	}
	err = out.Sync()
	return
}

// cleanupDir removes underlying files in path modified before specified time
func (s *Worker) cleanupDir(path string, before time.Time) {
	log.Info("cleaning %s < %s", path, before)
	files, err := ioutil.ReadDir(path)
	if err != nil {
		log.Warning("could not read %s err: %s", path, err.Error())
		return
	}

	oldFiles := make([]os.FileInfo, 0)
	// mark old file recursively
	for _, f := range files {
		if f.ModTime().Before(before) {
			oldFiles = append(oldFiles, f)
		}
		if f.IsDir() {
			s.cleanupDir(filepath.Join(path, f.Name()), before)
		}
	}

	for _, f := range oldFiles {
		fn := filepath.Join(path, f.Name())
		childFiles, _ := ioutil.ReadDir(fn)
		if !f.IsDir() || len(childFiles) == 0 {
			log.Info("removing %s", fn)
			if e := os.RemoveAll(fn); e != nil {
				log.Warning("could not remove %s err: %s", fn, e.Error())
			}
		}
	}
}
